package cn.com.cennavi.localModel;

import java.util.List;
import java.util.regex.Pattern;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.api.java.JavaStreamingContext;

import cn.com.cennavi.bean.GpsOrigInfo;
import scala.Tuple2;

/**
 * 
 * @ClassName: App
 * @Description: TODO（这里用一句话描述这个类的作用）
 * @author bull
 * @date 2016年3月14日 下午3:02:00
 *
 */
public class App {
	private static final Pattern SPACE = Pattern.compile(",");

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		/**
		 * 本地模式
		 */
		String logFile = "file:///D://data//DifPointTotalMapInfo.csv";
		SparkConf conf = new SparkConf().setAppName("testLocal").setMaster("local");
		JavaSparkContext sc = new JavaSparkContext(conf);
		JavaRDD<String> lines = sc.textFile(logFile).cache();
		testJoin();
		JavaPairRDD<String, String> ones = lines.mapToPair(new PairFunction<String, String, String>() {

			/**
			 * @Fields serialVersionUID : TODO（用一句话描述这个变量表示什么）
			 */
			private static final long serialVersionUID = 1L;

			@Override
			public Tuple2<String, String> call(String line) {
				String[] arr = line.split(",");
				// System.out.println(line);
				GpsOrigInfo goi = new GpsOrigInfo().getGpsOrigInfo(line);
				String key = goi.roadId + "_" + goi.dir;
				String value = line;
				return new Tuple2<String, String>(key, value);
			}
		});
		JavaPairRDD<String, String> counts = ones.reduceByKey(new Function2<String, String, String>() {

			/**
			 * @Fields serialVersionUID : TODO（用一句话描述这个变量表示什么）
			 */
			private static final long serialVersionUID = 1L;

			@Override
			public String call(String i1, String i2) {
				// System.out.println(i1);
				// System.out.println(i2);
				GpsOrigInfo goi1 = new GpsOrigInfo().getGpsOrigInfo(i1);
				String key1 = goi1.roadId + "_" + goi1.dir;
				// GpsOrigInfo goi2 = new GpsOrigInfo().getGpsOrigInfo(i2);
				// String key2 = goi2.roadId + "_" + goi2.dir;
//				System.out.println("key1------" + key1);
				// System.out.println("key2------" + key2);
				return i1 + i2;
			}
		}, 100);

		List<Tuple2<String, String>> output = counts.collect();
		for (Tuple2<?, ?> tuple : output) {
//			System.out.println(tuple._1() + ": " + tuple._2());
		}
		sc.stop();
	}

	private static void testJoin() {
		// TODO Auto-generated method stub
	}

}
